package com.yl.flink.processor;

import cn.hutool.core.util.NumberUtil;
import com.yl.entity.MultiDataEntity;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.List;
import java.util.Map;

/**
 * @author wlf
 * @since 2022/8/26
 */
public class MeanReduceFunc implements ReduceFunction<MultiDataEntity> {

    // 计算所用的指标序号
    private final int qc;

    public MeanReduceFunc(List<String> calRefers) {
        this.qc = Integer.parseInt(calRefers.get(0));
    }

    /**
     * 获取最大值
     * 注意：以窗口起始时间作为数据时间
     *
     * @param e1 前一个流元素
     * @param e2 后一个流元素
     */
    @Override
    public MultiDataEntity reduce(MultiDataEntity e1, MultiDataEntity e2) throws Exception {
        // 计算平均值
        double mean = NumberUtil.round((e1.getRawValues().get(qc) + e2.getRawValues().get(qc)) / 2, 3).doubleValue();
        // 以窗口起始时间作为数据时间
        e1.setCalValues(Tuple2.of(e1.getCalValues().f0, mean));
        return e1;
    }
}
